/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.core.redis;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class DataBuffer {

	private static class DataBlock {
		public DataBlock next;
		public DataBlock prev;
		public byte[] data;

		public DataBlock(byte[] data) {
			this.data = data;
			this.next = null;
			this.prev = null;
		}
	}

	private DataBlock head = null;
	private DataBlock tail = null;
	private DataBlock currentBlock = null;
	private DataBlock baseBlock = null;
	private int currentOffset = 0;
	private int baseOffset = 0;

	public DataBuffer() {
	}

	public void addData(byte[] data) {
		DataBlock block = new DataBlock(data);
		if (head == null) {
			assert (tail == null);
			assert (currentBlock == null);
			assert (baseBlock == null);
			head = block;
			tail = block;
			currentBlock = block;
			baseBlock = block;
			currentOffset = 0;
			baseOffset = 0;
		} else {
			assert (tail != null);
			assert (currentBlock != null);
			assert (baseBlock != null);
			tail.next = block;
			block.prev = tail;
			block.next = null;
			tail = block;
		}
	}

	public String readLine() {

		StringBuffer sb = new StringBuffer();
		Byte last = null;

		while (currentBlock != null) {
			byte[] data = currentBlock.data;
			for (; currentOffset < data.length; currentOffset++) {

				if (data[currentOffset] == '\n' && last != null
				        && ((char) last.byteValue()) == '\r') {
					currentOffset++;
					return sb.toString();
				}
				
				if (last != null) {
					sb.append((char)last.byteValue());
				}
				
				last = data[currentOffset];
			}
			
			currentBlock = currentBlock.next;
			currentOffset = 0;
		}
		
		return null;
	}

	public byte[] readBytes(int len) {
		
		byte[] ret = new byte[len];
		int offset = 0;
		int total = len;
		
		while (currentBlock != null) {
			int remaining = currentBlock.data.length - currentOffset;
			if (total <= remaining) {
				System.arraycopy(currentBlock.data, currentOffset, ret, offset, total);
				currentOffset += total;
				return ret;
			} else {
				System.arraycopy(currentBlock.data, currentOffset, ret, offset, remaining);
				offset += remaining;
				total -= remaining;
				currentOffset = 0;
				currentBlock = currentBlock.next;
			}
		}
		return null;
	}

	public void rollback() {
		currentBlock = baseBlock;
		currentOffset = baseOffset;
	}

	public void complete() {
		if (currentBlock == null) {
			head = null;
			tail = null;
			currentBlock = null;
			baseBlock = null;
			currentOffset = 0;
			baseOffset = 0;
		} else {
			baseBlock = currentBlock;
			baseOffset = currentOffset;
			head = baseBlock;
			if (baseBlock.prev != null)
				baseBlock.prev.next = null;
			baseBlock.prev = null;
		}
	}
	
	public static void main(String[] args) {
		
		DataBuffer buffer = new DataBuffer();
		
		buffer.addData("line1\r\nline".getBytes());
		buffer.addData("2\r\nline3\n\rabcd\r\nline4\r".getBytes());
		buffer.addData("\nline5\r".getBytes());
		
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		byte[] b = buffer.readBytes(7);
		System.out.println((b == null) ? "null" : new String(b));
		
		
		buffer.rollback();
		System.out.println("rollback1");
		
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		buffer.complete();
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		b = buffer.readBytes(7);
		System.out.println((b == null) ? "null" : new String(b));
		
		System.out.println("rollback2");
		buffer.rollback();
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		b = buffer.readBytes(7);
		System.out.println((b == null) ? "null" : new String(b));
		
		System.out.println("rollback3");
		buffer.addData("\nline6\r\n".getBytes());
		buffer.rollback();
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		System.out.println(buffer.readLine());
		buffer.complete();
		
		System.out.println("rollback4");
		System.out.println(buffer.readLine());
		
		
	}
}
